package com.example.springbootrocketmq.retry;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;

import java.io.UnsupportedEncodingException;

import static org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
import static org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus.RECONSUME_LATER;

/**
 * @Auther: 郭敬仰
 * @Date: 2019/8/24 22:39
 * @Description:消费端重试、异常重试
 */
public class ConsumerDemo {
    public static void main(String[] args) throws Exception {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("HAOKE_IM");
        consumer.setNamesrvAddr("192.168.64.128:9876");
       // 订阅topic，接收此Topic下的所有消息
        consumer.subscribe("my-test-topic", "*");
        consumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> {
            for (MessageExt msg : msgs) {
                try {
                    System.out.println(new String(msg.getBody(), "UTF-8"));
                } catch (UnsupportedEncodingException e) {
                    e.printStackTrace();
                }
            }
            System.out.println("收到消息->" + msgs);
            if (msgs.get(0).getReconsumeTimes() >= 3) {
               // 重试3次后，不再进行重试
                return CONSUME_SUCCESS;
            }
            return RECONSUME_LATER;
        });
        consumer.start();
    }
}
